package com.ruoyi.kafka.config;

import org.apache.kafka.common.TopicPartition;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.DeadLetterPublishingRecoverer;
import org.springframework.kafka.listener.DefaultErrorHandler;
import org.springframework.util.backoff.FixedBackOff;

/**
 * @ClassName KafkaErrorHandlerConfig
 * @Description Kafka 的统一异常处理配置
 * 消费者抛出异常 → 触发 DefaultErrorHandler。
 * 进入重试逻辑 →
 * 最多重试 3 次，每次间隔 1s。
 * 如果在重试过程中消费成功 → 正常提交 offset，不进入 DLT。
 * 重试 3 次仍失败 → 使用 DeadLetterPublishingRecoverer 把消息投递到 orders.DLT（对应分区）。
 * @Author li yang
 * @Date 2025/9/18 16:34
 * @Version 1.0
 */
@Configuration
public class KafkaErrorHandlerConfig {

    @Bean
    public DefaultErrorHandler errorHandler(KafkaTemplate<?, ?> template) {
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(
                template,
                (record, ex) -> new TopicPartition(record.topic() + ".DLT", record.partition())
        );

        // 重试 3 次，每次间隔 1s；超过则投递到 DLT
        return new DefaultErrorHandler(recoverer, new FixedBackOff(1000L, 3L));
    }

}
